package com.duoduo.demo.rabbitmq.rpc;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import com.rabbitmq.client.ConnectionFactory;

@Configuration
public class RabbitmqConfig {

	@Value("${spring.rabbitmq.host}")
	private String rmqHost = "172.16.185.233";
	@Value("${spring.rabbitmq.port}")
	private String rmqPort = "5672";
	@Value("${spring.rabbitmq.username}")
	private String rmqUsername = "dev";
	@Value("${spring.rabbitmq.password}")
	private String rmqPassword = "dev";

	@Value("${rmq.rpc.exchange}")
	private String rpcExchangeName = "demo.rpc.ex";
	@Value("${rmq.rpc.routingkey}")
	private String rpcRoutingKey = "demo.rpc.k";
	@Value("${rmq.rpc.requestqueue}")
	private String rpcRequestQueue = "demo.rpc.q";
	@Value("${rmq.rpc.replyqueue}")
	private String rpcReplyQueue = "demo.rpc.r";

	@Bean
	public ConnectionFactory connectionFactory() {
		ConnectionFactory connectionFactory = new ConnectionFactory();
		connectionFactory.setHost(rmqHost);
		connectionFactory.setPort(Integer.parseInt(rmqPort));
		// connectionFactory.setVirtualHost(env.getProperty("mq.vhost").trim());
		connectionFactory.setUsername(rmqUsername);
		connectionFactory.setPassword(rmqPassword);
		// ExecutorService service= Executors.newFixedThreadPool(20);//500个线程的线程池
		// connectionFactory.setSharedExecutor(service);
		return connectionFactory;
	}

	@Bean
	public CachingConnectionFactory rabbitConnectionFactory() {
		CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory(connectionFactory());
		cachingConnectionFactory.setPublisherConfirms(true);
		return cachingConnectionFactory;
	}

	// 定义rabbitmqTemplate
	@Bean
	public RabbitTemplate requestRabbitTemplate() {
		RabbitTemplate template = new RabbitTemplate(rabbitConnectionFactory());
		template.setExchange(rpcExchangeName);
		template.setRoutingKey(rpcRoutingKey);
		template.setQueue(rpcRequestQueue);
		// template.setReplyAddress(rpcReplyQueue);
		// template.setReplyAddress(requestExchange().getName() + "/" + replyQueue().getName());
		// template.setReceiveTimeout(60000);
		template.setReplyTimeout(60000);
		return template;
	}

	@Bean
	public Queue requestQueue() {
		return new Queue(rpcRequestQueue);
	}

	@Bean
	public TopicExchange requestExchange() {
		return new TopicExchange(rpcExchangeName);
	}

	@Bean
	public Binding bindingRequest() {
		return BindingBuilder.bind(requestQueue()).to(requestExchange()).with(rpcRoutingKey);
	}

	// 定义rabbitmqTemplate
	@Bean
	public RabbitTemplate replyRabbitTemplate() {
		RabbitTemplate template = new RabbitTemplate(rabbitConnectionFactory());
		template.setExchange(""); // 使用默认exchange
		template.setRoutingKey(rpcReplyQueue);
		// template.setReplyAddress(requestExchange().getName() + "/" + replyQueue().getName());
		// template.setReceiveTimeout(60000);
		template.setReplyTimeout(60000);
		return template;
	}

	@Bean
	public Queue replyQueue() {
		return new Queue(rpcReplyQueue);
	}

	// rabbitmqTemplate监听返回队列的消息
	@Bean
	public SimpleMessageListenerContainer replyListenerContainer() {
		SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
		container.setConnectionFactory(rabbitConnectionFactory());
		container.setQueues(replyQueue());
		container.setMessageListener(replyRabbitTemplate());
		ExecutorService executorService = Executors.newFixedThreadPool(300); // 300个线程的线程池
		container.setTaskExecutor(executorService);
		container.setConcurrentConsumers(200);
		container.setPrefetchCount(5);
		return container;
	}

	@Bean
	public RabbitAdmin admin() {
		return new RabbitAdmin(rabbitConnectionFactory());
	}
}
